package com.lmm.rpc.core.register;

import com.google.gson.Gson;
import com.lmm.rpc.core.cluster.RemoteServerInfo;
import com.lmm.rpc.core.config.ZKServerConfig;
import com.lmm.rpc.core.exception.RpcException;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.springframework.util.StringUtils;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;

@Slf4j
public class ZkRegister implements Registry {
    public static final int RETRY_TIMES = 3;
    public static final int SESSION_TIMEOUT = 3000;
    public static final String UTF_8 = "UTF-8";
    private static final int ZK_CONNECT_TIMEOUT = 5000;
    /**
     * zk配置文件
     * */
    private ZKServerConfig zkServerConfig;

    public CuratorFramework zkClient;

    public ZkRegister(ZKServerConfig zkServerConfig){
        if (zkServerConfig == null)
            throw new IllegalArgumentException("zkServerConfig can't be null！");
        this.zkServerConfig = zkServerConfig;
        init();
    }

    public void init(){
        //todo 参数校验
        String zkPath = zkServerConfig.getZkPath ();

        ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, RETRY_TIMES);
        zkClient = CuratorFrameworkFactory.builder()   //zkPath注册中心的地址
                //"ip:2181,ip2:2181,ip3:2181"
                .connectString(zkPath)
                .retryPolicy(retryPolicy)
                .connectionTimeoutMs(ZK_CONNECT_TIMEOUT)
                .sessionTimeoutMs(SESSION_TIMEOUT)
                .build();
        zkClient.start();
        try {
            //同步连接zk
            zkClient.blockUntilConnected(10, TimeUnit.SECONDS);
            log.info("zk server connect success!");
        } catch (InterruptedException e) {
            log.error("zk server connect fail,reason：【{}】",e.getMessage());
            throw new RuntimeException(e);
        }
        log.info("zk server init success,info={}", zkServerConfig);
    }

    /**
     * 注册方法
     * 创建一个接口节点信息
     * */
    public void registry(RemoteServerInfo serverInfo){
        String env = zkServerConfig.getEnv ();
        int port = zkServerConfig.getPort ();
//        int weight = serverInfo.getWeight ();
        String serviceInterfaceName = serverInfo.getServiceName();

//        String server = zkServerConfig.getServer ();
        if (zkClient == null){
            //（1）检查是否初始化
            log.error("zk server is null service:"+env+"-"+serviceInterfaceName);
            throw new IllegalArgumentException ( "zk server can't be null" );
        }
        try {
            //（2）检查对应zk节点是否创建
            String envPath = env.startsWith ( "/" ) ? env : "/".concat ( env );
            if (zkClient.checkExists().forPath(envPath) == null){
                zkClient.create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.PERSISTENT)
                        .forPath(envPath);
            }
            String servicePath = serviceInterfaceName.startsWith ( "/" ) ? serviceInterfaceName : "/".concat ( serviceInterfaceName );
            if (zkClient.checkExists().forPath(envPath+servicePath) == null){
                zkClient.create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.PERSISTENT)
                        //权限
                        .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                        .forPath(envPath+servicePath);
            }
            String ip = serverInfo.getIp();
            if (StringUtils.isEmpty ( ip )) {
                throw new IllegalArgumentException ( "ip can't be null" );
            }
            //（3）把封装的元信息转成json字符串
            Gson gson = new Gson();
            String jsonDataOfChild = gson.toJson(serverInfo);
            //（4）创建临时节点
            String childPath = envPath + servicePath + "/" + ip + ":" + port;
            if (zkClient.checkExists().forPath(childPath) == null){
                zkClient.create()
                        .creatingParentsIfNeeded()
                        //临时节点
                        .withMode(CreateMode.EPHEMERAL)
                        .forPath(childPath,jsonDataOfChild.getBytes(StandardCharsets.UTF_8));
            }
        }catch (Exception e){
            log.error(e.getMessage(),e);
        }
    }

    @Override
    public void destroy() {
        String env = zkServerConfig.getEnv();
        try {
            zkClient.delete().forPath(env.startsWith ( "/" ) ? env : "/".concat ( env ));
        } catch (Exception e) {
            log.error("env:"+env+" remove fail!");
            throw new RpcException(e);
        }finally {
            zkClient.close();
        }
    }
}
